package function;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Long>> {

    private transient ValueState<Tuple2<Long,Long>> sum;
    @Override
    public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {

        Tuple2<Long,Long> currentSum =sum.value();
        currentSum.f0+=1;
        // update the state
        sum.update(currentSum);


    }
}
